package com.tensai.study.java.newconcept.typereactiveprogram.reactor.disposable;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.Disposable;
import reactor.core.Disposables;
import reactor.core.publisher.Flux;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author Tensai
 */
@Slf4j
public class SwapTest {

	/**
	 * 停止上一个流切换到下一个流
	 *
	 * @throws InterruptedException 线程异常
	 */
	@Test
	public void updateTest() throws InterruptedException {
		Flux<String> flux = Flux.interval(Duration.ofMillis(400)).map(Object::toString);

		Disposable.Swap swap = Disposables.swap();
		for (int i = 0; i < 10; i++) {
			swap.update(flux.subscribe(log::info));
			TimeUnit.SECONDS.sleep(2);
		}
	}

	@Test
	public void replaceTest() throws InterruptedException {
		Flux<Long> flux = Flux.interval(Duration.ofMillis(400));

		Disposable.Swap swap = Disposables.swap();
		for (double i = 0.1; i < 0.7; i = i + 0.1) {
			double finalI = i;
			swap.replace(flux.map(it -> it + finalI / 10).map(Object::toString).subscribe(log::info));
			TimeUnit.SECONDS.sleep(2);
		}

		TimeUnit.MINUTES.sleep(1);
	}
}
